package com.hzlj.position.dbsync.liaoning.service.impl;

import cn.hutool.core.util.ObjectUtil;
import com.fz.common.base.dto.PageParamDTO;
import com.fz.common.base.dto.QueryPageResultDTO;
import com.fz.common.base.dto.SortParamDTO;
import com.fz.common.base.exception.PubException;
import com.fz.common.dict.annotation.DictConvert;
import com.hzlj.position.dbsync.common.service.SyncOffsetConfigService;
import com.hzlj.position.dbsync.liaoning.common.dto.messageSync.MessageSyncQueryParamDTO;
import com.hzlj.position.dbsync.liaoning.common.dto.messageSync.MessageSyncQueryResultDTO;
import com.hzlj.position.dbsync.liaoning.common.enums.MessageSyncType;
import com.hzlj.position.dbsync.liaoning.common.model.MessageSync;
import com.hzlj.position.dbsync.liaoning.dao.MessageSyncDao;
import com.hzlj.position.dbsync.liaoning.service.MessageSyncService;
import com.hzlj.position.locate.common.common.ReadUpdateDTO;
import com.hzlj.position.locate.service.PositionService;
import com.hzlj.position.locate.service.ReportInspectionService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.Date;
import java.util.List;

/**
 * 消息已读记录-用来做消息同步使用(MessageSync)表服务接口
 *
 * @author lifh
 * @date 2023-11-30 09:13:04
 */
@Slf4j
@Service
public class MessageSyncServiceImpl implements MessageSyncService {
    private final static String SYNC_JG_CODE = "1";
    private final static String SYNC_TYPE = "message_read_sync";
    private final static Integer SYNC_SIZE = 500;

    @Resource
    private MessageSyncDao messageSyncDao;
    @Resource
    private SyncOffsetConfigService syncOffsetConfigService;
    @Resource
    private PositionService positionService;
    @Resource
    private ReportInspectionService reportInspectionService;

    /**
     * 查询列表：不分页
     */
    @Override
    @DictConvert
    public QueryPageResultDTO<MessageSyncQueryResultDTO> pageMessageSync(MessageSyncQueryParamDTO dto, PageParamDTO page) {
        return this.messageSyncDao.pageMessageSync(dto, page);
    }

    private List<MessageSyncQueryResultDTO> listMessageSync() {
        PageParamDTO pageParamDTO = new PageParamDTO();
        pageParamDTO.setSize(SYNC_SIZE);
        pageParamDTO.setSortFns(SortParamDTO.of(true, MessageSync::getReadTime).getSortFns());
        //最近同步时间
        Date existsReadTime = syncOffsetConfigService.getLatestTime(SYNC_JG_CODE, SYNC_TYPE);
        //分页查询
        QueryPageResultDTO<MessageSyncQueryResultDTO> latestPageRecords = pageMessageSync(
                new MessageSyncQueryParamDTO().setStartReadTime(existsReadTime), pageParamDTO
        );
        return latestPageRecords.getList();
    }

    @Override
    public long sync() {
        long size = syncOneStep();
        //需要继续同步
        if (size - SYNC_SIZE >= 0) {
            size += syncOneStep();
        }
        return size;
    }

    private long syncOneStep() {
        List<MessageSyncQueryResultDTO> latestRecords = listMessageSync();
        //同步记录
        for (MessageSyncQueryResultDTO record : latestRecords) {
            try {
                sync(record);
            } catch (Exception e) {
                log.error("同步消息出错:{}", record, e);
            }
        }

        if (ObjectUtil.isEmpty(latestRecords)) {
            return 0L;
        }

        //同步时间
        syncOffsetConfigService.upsert(SYNC_JG_CODE, SYNC_TYPE, latestRecords.get(latestRecords.size() - 1).getReadTime());
        return latestRecords.size();
    }

    private void sync(MessageSyncQueryResultDTO message) {
        MessageSyncType type = MessageSyncType.of(message.getRecordType());
        log.info("开始同步消息:{}", message);
        ReadUpdateDTO update = new ReadUpdateDTO()
                .setId(message.getId())
                .setReadTime(message.getReadTime())
                .setRecordTime(message.getRecordTime());

        switch (type) {
            case INSPECTION:
                reportInspectionService.read(update);
                break;
            case POSITION:
                positionService.read(update);
                break;
            default:
                log.warn("未定义消息同步类型:{}", message);
                throw new PubException("未定义消息同步类型:" + message.getRecordType());
        }
    }
}
